package com.flink.datasource;

import com.flink.entity.User;
import com.sequence.IdUtil;
import com.sequence.Sequence;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;

/**
 * 使用忙循环模拟一个无界流
 * 两个方法 run负责执行；cancel负责停止 记得给一个状态位
 * TODO 多线程ParallelSourceFunction 并行度可以设置多个
 */
public class UserSource implements SourceFunction<User> {
    volatile boolean isRunning = true;
    static List<String> list = new ArrayList<>();
    private Integer sleep = 0;

    public UserSource(Integer sleep) {
        this.sleep = sleep;
    }

    public UserSource() {
    }

    static {
        list.add("/家电");
        list.add("/零食");
        list.add("/洗发水");
        list.add("/椅子");
        list.add("/手机");
        list.add("/电脑");
        list.add("/音响");
        list.add("/雪糕");
    }

    private static final ThreadLocalRandom RANDOM = ThreadLocalRandom.current();
    //private static final ThreadLocalRandom RANDOM2 = ThreadLocalRandom.current();

    @Override
    public void run(SourceContext<User> ctx) throws Exception {
        while (isRunning) {
            User user=new User();
            Long timestamp = Calendar.getInstance().getTimeInMillis();
            int r = RANDOM.nextInt(8);
            int i = RANDOM.nextInt(10);
            //user.setId(IdUtil.getId());
            user.setTimestamp(timestamp);
            user.setProd(list.get(r));
            user.setStatus(RANDOM.nextInt(2));
            user.setName( "用户"+i);
            ctx.collect(user);
            //慢点执行可以看数据比较直观
            if (sleep!=0)
            Thread.sleep(sleep);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}

